package com.example.lightenergypolymerize.controller;

import cn.hutool.core.bean.BeanUtil;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.example.lightenergypolymerize.entity.Discuss;
import com.example.lightenergypolymerize.entity.DisscussSupport;
import com.example.lightenergypolymerize.entity.User;
import com.example.lightenergypolymerize.entity.vo.CommentVO;
import com.example.lightenergypolymerize.entity.vo.DiscussVO;
import com.example.lightenergypolymerize.service.ICommentService;
import com.example.lightenergypolymerize.service.IDiscussService;
import com.example.lightenergypolymerize.service.IDisscussSupportService;
import com.example.lightenergypolymerize.service.IUserService;
import com.example.lightenergypolymerize.util.AppVariable;
import com.example.lightenergypolymerize.util.ResponseEntity;
import com.example.lightenergypolymerize.util.SecurityUtil;
import io.reactivex.rxjava3.core.Single;
import jakarta.annotation.Resource;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.validation.annotation.Validated;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.HashMap;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;

@RestController
@RequestMapping("/discuss")
public class DiscussController {
    @Resource
    private IDiscussService discussService;
    @Resource
    private ThreadPoolTaskExecutor threadPool;
    @Resource
    private IUserService userService;
    @Resource
    private ICommentService commentService;
    @Resource
    private KafkaTemplate kafkaTemplate;
    @Resource
    private IDisscussSupportService disscussSupportService;
    /**
     * 添加话题讨论
     */
    @RequestMapping("/add")
    public ResponseEntity add(@Validated Discuss discuss) {
        discuss.setUid(SecurityUtil.getCurrentUser().getUid());
        boolean save = discussService.save(discuss);
        if(save) {
            return ResponseEntity.succ(save);
        }
        return ResponseEntity.fail("保存失败，请重试");
    }


    /**
     * 获取我的话题讨论列表
     */
    @RequestMapping("/mylist")
    public ResponseEntity getMyList() {
        return ResponseEntity.succ(discussService.list(
                Wrappers.lambdaQuery(Discuss.class)
                        .eq(Discuss::getUid, SecurityUtil.getCurrentUser().getUid())
                        .orderByDesc(Discuss::getDid)
        ));
    }


    /**
     * 删除我的话题讨论
     */
    @RequestMapping("/delete")
    public ResponseEntity delete(Long did) {
        if(did == null || did <= 0) {
            return ResponseEntity.fail("参数错误");
        }
        boolean result = discussService.remove(
                Wrappers.lambdaQuery(Discuss.class)
                        .eq(Discuss::getDid, did)
                        .eq(Discuss::getUid, SecurityUtil.getCurrentUser().getUid())
        );
        if(result) {
            return ResponseEntity.succ(result);
        }
        return ResponseEntity.fail("删除失败，请重试");
    }




    /**
     * 查询话题讨论详情
     *
     */
    @RequestMapping("/detail")
    public ResponseEntity detail(Long did) throws ExecutionException, InterruptedException {
        if(did == null || did <= 0) {
            return ResponseEntity.fail("参数错误");
        }
        Discuss discuss = discussService.getById(did);
        if(discuss != null && discuss.getDid() > 0) {
            //阅读量加1，不要将数据取出来加1再放回数据库；
            //要用mysql的update语句，因为mysql的update语句在mysql层面上是线程安全的。
            //不需要线程编排
            threadPool.submit(() -> {
               //1.更新数据
               discussService.updateReadCount(did);
               //2.返回对象阅读数+1
                discuss.setReadcount(discuss.getReadcount() + 1);
            });


            //任务1.查询discuss中的username
            CompletableFuture<DiscussVO> task1 = CompletableFuture.supplyAsync(() -> {
                //对象转换
                DiscussVO discussVO = BeanUtil.toBean(discuss, DiscussVO.class);
                User user = userService.getById(discuss.getUid());
                if(user != null && user.getUid() > 0) {
                    discussVO.setUsername(user.getUsername());
                }
                return discussVO;
            }, threadPool);

            //任务2.查询discuss所对应的comment列表
            CompletableFuture<List<CommentVO>> task2 = CompletableFuture.supplyAsync(() -> {
                return commentService.getCommentList(did);
            }, threadPool);
            //任务1和任务2执行完了才能继续向下执行，所以需要线程编排
            CompletableFuture.allOf(task1, task2);
            HashMap<String, Object> result = new HashMap<>();
            result.put("discuss", task1.get());
            result.put("commentlist", task2.get());
            return ResponseEntity.succ(result);
        }
        return ResponseEntity.fail("该话题讨论不存在");
    }


    /**
     * 话题点赞
     * 点赞需要写讨论表和点赞表两张表，要保证原子性。在微服务中单体事务不起作用，所以引入消息队列。
     * 先将点赞事件写入mq，然后后面再消费mq中的点赞事件，等讨论表和点赞表两张表都写入了新内容后，再对mq进行手动确认。
     * 不管讨论表和点赞表两张表有没有都写入新内容，都要对mq进行手动确认，不然mq会有消息积压问题
     */
    @RequestMapping("/support")
    public ResponseEntity support(Long did) {
        if(did == null || did <= 0) {
            return ResponseEntity.fail("参数错误");
        }
        //将点赞事件写入mq
        kafkaTemplate.send(AppVariable.DISCUSS_SUPPORT_TOPIC, did + "_" +
                SecurityUtil.getCurrentUser().getUid());
        return ResponseEntity.succ(true);
    }


    /**
     * 监听mq中的点赞事件
     */
    //这里监听的是mq的上下文，而不是springsecurity的上下文，
    // 所以在下面这个方法里用SecurityUtil.getCurrentUser().getUid()会有问题
    //下面这个方法不要用SecurityUtil.getCurrentUser().getUid()
    @KafkaListener(topics = AppVariable.DISCUSS_SUPPORT_TOPIC)
    public void listen(String data, Acknowledgment acknowledgment) {
        //1.判断当前登录用户是否有给当前话题点赞过，要查点赞表，如果点赞过则不能继续点赞
        Long did = Long.parseLong(data.split("_")[0]);
        Long uid = Long.parseLong(data.split("_")[1]);
        List<DisscussSupport> list = disscussSupportService.list(
                Wrappers.lambdaQuery(DisscussSupport.class)
                        .eq(DisscussSupport::getDid, did)
                        .eq(DisscussSupport::getUid, uid)
        );
        if(list == null || list.size() == 0) {
            //2.修改讨论表点赞数
            int result = discussService.updateSupportCount(did);
            if (result > 0) {
                //3.在点赞表添加点赞信息
                DisscussSupport disscussSupport = new DisscussSupport();
                disscussSupport.setDid(did);
                disscussSupport.setUid(uid);
                disscussSupportService.save(disscussSupport);
            }
        }
        //4.手动确认消息已消费
        acknowledgment.acknowledge();
    }


    /**
     * 获取话题列表（有分页功能）
     * page是第几页
     * type:1=推荐（点赞数排序）2=最新（时间排序）
     */
    @RequestMapping("/list")
    public ResponseEntity list(Integer page, Integer type) {
        if(page == null || page <= 0) {
            page = 1;
        }
        if(type == null || type <= 0) {
            type = 1;
        }
        QueryWrapper<Discuss> queryWrapper  =new QueryWrapper<>();
        if(type == 1) {//根据点赞数进行排序
            queryWrapper.orderByDesc("supportcount");
        }else {//根据创建时间排序
            //这里不用时间进行倒叙排序，因为时间没有索引，查询速度慢
            // 用了主键进行倒叙排序，是因为主键和时间的顺序是一样的，用主键代替时间
            queryWrapper.orderByDesc("did");
        }
        Page<Discuss> result = discussService.page(new Page<>(page, AppVariable.PAGE_SIZE),
                queryWrapper
        );
        return ResponseEntity.succ(result);
    }
}


